package tpc;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.SocketException;
import java.util.ArrayList;

import commsCPN.JavaCPN;

public class Coordinator {
    private static Integer numOfWorkers = null;
    private static JavaCPN jCPN = new JavaCPN();
    private static STATES state = STATES.INIT;

    public enum STATES {
        INIT, RECEIVE_DECISIONS, RECEIVE_ACKS;
        public STATES getNext() {
            return this.ordinal() < STATES.values().length - 1 ? STATES.values()[this.ordinal() + 1] : STATES.INIT;
        }
    };

    public static void main(String[] args) throws IOException, InterruptedException {
        System.out.println("Coordinator running.");
        String[] received = null;
        CollectVotes cV = null;
        int expectedAcks = 0;
        while (true) {
            switch (state) {
            case INIT:
                try {
                    jCPN.connect("localhost", 9000);
                } catch (java.net.ConnectException e) {
                    Thread.sleep(100);
                    break;
                }
                received = receive();
                if (received[0].equals("numOfWorkers")) {
                    state = state.getNext();
                    numOfWorkers = Integer.parseInt(received[received.length - 1]);
                    sendCanCommitMessages(numOfWorkers);
                }
                break;
            case RECEIVE_DECISIONS:
                cV = new CollectVotes(jCPN, numOfWorkers);
                expectedAcks = cV.run();
                state = state.getNext();
                break;
            case RECEIVE_ACKS:
                //receive the expected acks, reset state.
                while (expectedAcks>0) {
                    received = receive();
                    expectedAcks--;
                }
                jCPN.send(new ByteArrayInputStream("".getBytes()));
                state = state.getNext();
                jCPN.disconnect();
                System.out.println("Cycle completed.");
                break;
            }
        }
    }

    private static void sendCanCommitMessages(int numOfWorkers) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < numOfWorkers; i++)
            sb.append(i).append(';');
        try {
            System.out.println("Sending: " + sb.toString());
            jCPN.send(new ByteArrayInputStream(sb.toString().getBytes()));
        } catch (SocketException e) {
            e.printStackTrace();
        }
    }

    private static String[] receive() {
        String received = null;
        try {
            received = jCPN.receive().toString();
        } catch (java.net.SocketException e) {
            state = STATES.INIT;
        }
        System.out.println("Received data: " + received);
        return received.split(";");
    }
}
